package com.zhang.app.func;

import com.alibaba.fastjson.JSONObject;
import com.zhang.common.GmallConfig;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

/**
 * @title: 自定义JDBC Sink
 * @author: zhang
 * @date: 2022/5/5 11:06
 */
public class DimSinkFunction extends RichSinkFunction<JSONObject> {
    private Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
        connection.setAutoCommit(true);
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        PreparedStatement statement = null;
        try {
            String sinkTable = value.getString("sinkTable");
            JSONObject data = value.getJSONObject("data");
            // 拼接sql
            String upsertSQL = genUpsertSql(sinkTable, data);
            System.out.println("upsert语句：" + upsertSQL);

            statement = connection.prepareStatement(upsertSQL);
            statement.executeUpdate();
        } catch (SQLException e) {
            System.out.println("插入数据失败！");
        } finally {
            if (statement != null) {
                statement.close();
            }
        }

    }

    // upsert into student(id,name,sex) values('1001','zhangsan','beijing');
    private String genUpsertSql(String sinkTable, JSONObject data) {
        Set<String> keySet = data.keySet();
        Collection<Object> values = data.values();
        StringBuilder upsertSQL = new StringBuilder("upsert into ")
                .append(GmallConfig.HBASE_SCHEMA)
                .append(".")
                .append(sinkTable)
                .append(" (")
                .append(StringUtils.join(keySet, ","))
                .append(") ")
                .append("values('")
                .append(StringUtils.join(values, "','"))
                .append("')");
        return upsertSQL.toString();
    }

    @Override
    public void close() throws Exception {
        connection.close();
    }
}
